package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

import java.lang

object Demo19Accumulator {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    conf.setMaster("local")
    conf.setAppName("app")

    val sc = new SparkContext(conf)

    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.csv")

    var sum: Int = 0
    val mapRDD: RDD[String] = studentsRDD
      .map(line => {
        //当再算子内使用算子外的一个普通变量是，这个变量会被封装成一个变量副本发送到Executor中
        //所以再编写Spark代码是不能再算子内去修改算子外的一个普通变量
        sum += 1
        println(s"sum:$sum")
        line
      })
    val num: Long = mapRDD.count()
    println(s"num:$num")
    println(s"sum:$sum")

    /**
     * 累加器
     * 1、先在每一个Executor总局部累加
     * 2、当job执行完成，将累加结果拉取到Driver端合并累加结果
     * 累加器不会触发任务执行，可以挂在任意的一个任务下，比count算子的效率要高
     */

    //定义累加器
    val accSum: LongAccumulator = sc.longAccumulator
    val rdd2: RDD[String] = studentsRDD
      .map(line => {
        //对累加器进行累加
        accSum.add(1)
        line
      })
    val count: Long = rdd2.count()
    //获取累加结果
    val value: Long = accSum.value
    println(s"accSum:$value")
    println(s"count:$count")

  }
}
